package com.bw.yk05;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class Mn5TM4_8 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.executeSql("create table mn5TM4Wide (\n" +
                "    city_name string,\n" +
                "    courier_name string,\n" +
                "    create_time string,\n" +
                "    times as to_timestamp(create_time),\n" +
                "    WATERMARK FOR times AS times - INTERVAL '0' SECOND\n" +
                ")with(\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'mn5tm4topic4',\n" +
                "    'properties.bootstrap.servers' = 'hadoop-single:9092',\n" +
                "    'properties.group.id' = 'group1',\n" +
                "    'scan.startup.mode' = 'earliest-offset',\n" +
                "    'format' = 'json'\n" +
                ")");
        Table table1 = tEnv.sqlQuery("" +
                "select " +
                " TUMBLE_START(times,INTERVAL '1' day) as wsStart," +
                " TUMBLE_END(times,INTERVAL '1' day) as wsEnd," +
                " city_name," +
                " count(1) cnts1 from mn5TM4Wide " +
                "GROUP BY TUMBLE(times,INTERVAL '1' day),city_name");
        tEnv.createTemporaryView("tmp1",table1);
        /*
        +----+-------------------------+-------------------------+--------------------------------+----------------------+
        | op |                 wsStart |                   wsEnd |                      city_name |                cnts1 |
        +----+-------------------------+-------------------------+--------------------------------+----------------------+
        | +I |        2023-07-12T00:00 |        2023-07-13T00:00 |                         长沙市 |                   12 |
        | +I |        2023-07-12T00:00 |        2023-07-13T00:00 |                         合肥市 |                    5 |
         */

        Table table2 = tEnv.sqlQuery("" +
                "select " +
                " TUMBLE_START(times,INTERVAL '1' day) as wsStart," +
                " TUMBLE_END(times,INTERVAL '1' day) as wsEnd," +
                " city_name,courier_name," +
                " count(1) cnts2 from mn5TM4Wide " +
                "GROUP BY TUMBLE(times,INTERVAL '1' day),city_name,courier_name");
        tEnv.createTemporaryView("tmp2",table2);

        /*
        +----+-------------------------+-------------------------+--------------------------------+--------------------------------+----------------------+
        | op |                 wsStart |                   wsEnd |                      city_name |                   courier_name |                cnts2 |
        +----+-------------------------+-------------------------+--------------------------------+--------------------------------+----------------------+
        | +I |        2023-07-12T00:00 |        2023-07-13T00:00 |                         长沙市 |                         彭亮政 |                    6 |
        | +I |        2023-07-12T00:00 |        2023-07-13T00:00 |                         合肥市 |                           秦倩 |                    5 |
        | +I |        2023-07-12T00:00 |        2023-07-13T00:00 |                         西安市 |                           姚龙 |                    4 |
     */
        tEnv.sqlQuery("select t1.wsStart,t1.wsEnd,t1.city_name,t2.courier_name,t1.cnts1,t2.cnts2,cast(t2.cnts2 as double)/t1.cnts1 " +
                " from tmp1 t1 join tmp2 t2" +
                " on t1.wsStart = t2.wsStart and  " +
                " t1.city_name = t2.city_name").execute().print();

    }

}














